package com.sikong.geomesa;
import com.sikong.configuration.Constant;
import org.geotools.data.DataStore;
import org.geotools.data.DataStoreFinder;
import org.geotools.data.FeatureWriter;
import org.geotools.data.Transaction;
import org.geotools.filter.identity.FeatureIdImpl;
import org.geotools.util.factory.Hints;

import org.locationtech.geomesa.utils.interop.SimpleFeatureTypes;
import org.opengis.feature.simple.SimpleFeature;
import org.opengis.feature.simple.SimpleFeatureType;

import java.io.IOException;
import java.util.HashMap;
import java.util.Map;

/**
 * 这是一段用于示范GeoMesa-HBase插入数据的逻辑代码
 * This is an example coding of GeoMesa-HBase Data Insertion/Ingestion（Officially called Ingestion from geomesa-hbase shell)
 * @author 梁超
 * @since 0.1.0
 * @Date 2022/03/03
 */
public class PutFeature {

    private FeatureWriter<SimpleFeatureType, SimpleFeature> writer;

    /**
     *此方法用于连接HBase
     * @param tableName 用于创建主表的名称
     * @return Datastore 用于写入数据的容器
     */
    public DataStore mkConnection(String tableName) throws IOException {
        Map<String,String> params = new HashMap<>();
        params.put(Constant.HBASE_CATALOG,tableName);
        params.put(Constant.ZK, Constant.ZK_HOST);

        return DataStoreFinder.getDataStore(params);
    }

    /**
     * 此方法用于构建表的类型，如第一列是String类型。
     * @param secondTableName Geomesa-HBase有索引副表，该名称在主表名称基础上添加
     * @param tableStructure column的类型结构，具体参考GeoTools
     * @return sft
     */
    public SimpleFeatureType setTable(String secondTableName, String tableStructure) {
        SimpleFeatureType sft = SimpleFeatureTypes.createType(secondTableName, tableStructure);
        sft.getUserData().put(SimpleFeatureTypes.DEFAULT_DATE_KEY,"dtg");

        return sft;
    }

    /**
     *
     * @param dataStore 同上
     * @param sft 构建好要素类
     * @param feature 构建好的要素数据
     */
    public void writeFeature(DataStore dataStore, SimpleFeatureType sft, SimpleFeature feature) throws IOException {
        this.writer = dataStore.getFeatureWriterAppend(sft.getTypeName(), Transaction.AUTO_COMMIT);
        SimpleFeature toWrite = writer.next();
        toWrite.setAttributes(feature.getAttributes());
        ((FeatureIdImpl) toWrite.getIdentifier()).setID(feature.getID());
        toWrite.getUserData().put(Hints.USE_PROVIDED_FID,Boolean.TRUE);
        toWrite.getUserData().putAll(feature.getUserData());
        writer.write();
    }

    public void writeClose() throws IOException {
        writer.close();
    }
}
